package com.bitcoin.scheduler.service;

import com.bitcoin.scheduler.domain.blockcc.KLineBlockCC;
import com.bitcoin.scheduler.repository.KLineBlockCCDao;
import com.bitcoin.scheduler.repository.MarketSymbolPairsDao;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.hibernate.query.NativeQuery;
import org.hibernate.transform.Transformers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.EntityManager;
import javax.persistence.Query;
import java.text.ParseException;
import java.time.Duration;
import java.time.Instant;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@Slf4j
@Service
public class BlockccService {
    @Autowired
    @Getter
    private MarketSymbolPairsDao symbolPairsDao;
    @Autowired
    @Getter
    private EntityManager entityManager;
    @Autowired
    @Getter
    private KLineBlockCCDao kLineBlockCCDao;
    @Autowired
    MarketBinanceApiService binanceApiService;
    @Autowired
    MarketGateIoApiService gateIoApiService;

    @Transactional
    public void removeByMarketAndSymbolAndPriceTimeBetween(String market, String symbol, Date startDate, Date endDate) {
        kLineBlockCCDao.removeByMarketAndSymbolAndPriceTimeBetween(market, symbol, startDate, endDate);
    }

    @Transactional
    public Future<Integer> sync(String market, String symbol, int limit) {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return CompletableFuture.supplyAsync(() -> syncData(market, symbol, limit));
    }

    @Transactional
    public Integer syncData(String market, String symbol, int limit) {
        long startTime = System.currentTimeMillis();
        List<KLineBlockCC> klineList = null;
        try {
            if (StringUtils.equals(MarketBinanceApiService.MARKET_NAME, market)) {
                klineList = binanceApiService.getKlineResponse(symbol, limit);
            }
            else {
                klineList = gateIoApiService.getKlineResponse(symbol, limit);
            }
        } catch (Exception e) {
            try {
                return sync(market, symbol, limit).get();
            } catch (InterruptedException | ExecutionException e1) {
                e1.printStackTrace();
            }
        }
        if (CollectionUtils.isEmpty(klineList)) {
            log.warn("{}/{} 无数据", market, symbol);
            try {
                Thread.sleep(3000);
                return sync(market, symbol, limit).get();
            } catch (Exception e) {
                return 0;
            }
        }
        Date startDate = klineList.get(0).getPriceTime();
        Date endDate = klineList.get(klineList.size() - 1).getPriceTime();
        log.info(" 市场{} 币{} 时间段{}-{}", market, symbol,
            DateFormatUtils.format(klineList.get(0).getPriceTime(), "yyyy/MM/dd HH:mm:ss"),
            DateFormatUtils.format(klineList.get(klineList.size() - 1).getPriceTime(), "yyyy/MM/dd HH:mm:ss"));
        //取得已有日期点的日期
        Map<String, Integer> pointsMap = getPoints(market, symbol, startDate, endDate);
        Map<String, List<KLineBlockCC>> catchPriceTimeMap = Maps.newHashMap();

        for (KLineBlockCC line : klineList) {
            String dayKey = DateFormatUtils.format(line.getPriceTime(), "yyyy/MM/dd");
            if (pointsMap.containsKey(dayKey) && pointsMap.get(dayKey) == 288) {
            }
            else {
                if (catchPriceTimeMap.containsKey(dayKey)) {
                    catchPriceTimeMap.get(dayKey).add(line);
                }
                else {
                    catchPriceTimeMap.put(dayKey, Lists.newArrayList(line));
                }
            }
        }
        for (String date : catchPriceTimeMap.keySet()) {
            catchPriceTimeMap.get(date).sort(Comparator.comparing(KLineBlockCC::getPriceTime));
            startDate = catchPriceTimeMap.get(date).get(0).getPriceTime();
            endDate = catchPriceTimeMap.get(date).get(catchPriceTimeMap.get(date).size() - 1).getPriceTime();
            removeByMarketAndSymbolAndPriceTimeBetween(market, symbol, startDate, endDate);
            getKLineBlockCCDao().saveAll(catchPriceTimeMap.get(date));
            getKLineBlockCCDao().flush();
        }
        log.info("数据保存完毕  交易市场 {} 币标识符{} 耗时 {}ms", market, symbol, System.currentTimeMillis() - startTime);
        return 1;
    }

    @Transactional
    protected Map<String, Integer> getPoints(String market, String symbol, Date startDate, Date endDate) {
        String sql = "" +
            "SELECT symbol, market, DATE_FORMAT(price_time, '%Y/%m/%d') date, count(1) points " +
            "FROM block_cc_5m_kline " +
            "WHERE symbol = :symbol " +
            "  AND market = :market " +
            "  AND price_time >= :startDate " +
            "  AND price_time <= :endDate " +
            "GROUP BY symbol, market, date" +
            "";
        Query query = getEntityManager().createNativeQuery(sql);
        query.setParameter("startDate", DateFormatUtils.format(startDate, "yyyy/MM/dd HH:mm:ss"));
        query.setParameter("endDate", DateFormatUtils.format(endDate, "yyyy/MM/dd HH:mm:ss"));
        query.setParameter("symbol", symbol);
        query.setParameter("market", market);
        query.unwrap(NativeQuery.class).setResultTransformer(Transformers.ALIAS_TO_ENTITY_MAP);
        List<Map> result = query.getResultList();
        Map<String, Integer> map = Maps.newConcurrentMap();
        for (Map entity : result) {
            map.put(MapUtils.getString(entity, "date"), MapUtils.getInteger(entity, "points"));
        }
        return map;
    }


    @Scheduled(fixedDelay = 6000)
    @Transactional
    public void syncUnFullData() throws ParseException {
        String sql = "" +
            "SELECT market,symbol,DATE_FORMAT(price_time,'%Y%m%d') day ,count(1) points FROM block_cc_5m_kline where price_time >= '2018/09/01' group by market,bcc_id,day having count(1) != 288 order by day asc ";
        Query query = getEntityManager().createNativeQuery(sql);
        query.unwrap(NativeQuery.class).setResultTransformer(Transformers.ALIAS_TO_ENTITY_MAP);
        List<Map> result = query.getResultList();
        for (Map entity : result) {
            Date date = DateUtils.parseDate(MapUtils.getString(entity, "day"), "yyyyMMdd");
            Long days = Duration.between(Instant.ofEpochSecond(date.getTime() / 1000), Instant.now()).toDays();

            syncData(MapUtils.getString(entity, "market"),MapUtils.getString(entity, "symbol"),days.intValue());
        }
    }
}
